package org.chnxi.study.messagequeue.kafka.comsumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class SimpleConsumer {

    public void consumer(){
        Properties props = new Properties();
        // kafka的服务地址
        props.setProperty("bootstrap.servers","");
        // 定义consumer所属的消费者组
        props.setProperty("group.id","consumerGroupName");
        // 开启自动确认offset
        props.setProperty("enable.auto.commit","true");
        // 自动确认offset的时间间隔，毫秒
        props.setProperty("auto.commit.interval.ms","1000");
        // 设置key和value的反序列化处理类
        props.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

        // 创建consumer对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);
        // 设置要消费的topic，可以设置多个
        kafkaConsumer.subscribe(Arrays.asList("topicName_1", "topicName_2"));

        while(true){
            // 读取数据，设置超时时间是100ms
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(100);
            consumerRecords.forEach(consumerRecord ->{
                // 从消息中获取topic的名称
                String topic = consumerRecord.topic();
                // 从消息中获取所在的partition
                int partition = consumerRecord.partition();
                // 从消息中获取当前的offset
                long offset = consumerRecord.offset();
                // 从消息中获取 key
                String key = consumerRecord.key();
                // 从消息中获取 value
                String value = consumerRecord.value();
            });
        }
    }

}
